package HandshakeJoin;

import org.apache.s4.base.Event;

public class Message extends Event
{
    public static enum Type {
        TUPLE, ACK, TUPLE_BLK, ACK_BLK, SIZE_CHG, UNKNOWN
    }

    public static enum Direction {
        FROM_LEFT, FROM_RIGHT, UNKNOWN
    }

    private String joinId = null;
    private Type type = Type.UNKNOWN;
    private Direction from = Direction.UNKNOWN;
    private int windowSize = 0;
    private Tuple tuple = null;
    private TupleBlock tupleBlk = null;

    Message(final String joinId, final Type type, final Direction from,
            final int windowSize) {
        this.joinId = joinId;
        this.type = type;
        this.from = from;
        this.windowSize = windowSize;
    }

    Message(final String joinId, final Direction from, final Tuple tuple) {
        this.joinId = joinId;
        type = Type.TUPLE;
        this.from = from;
        this.tuple = tuple;
    }

    Message(final String joinId, final Direction from, final TupleBlock tupleBlk) {
        this.joinId = joinId;
        type = Type.TUPLE_BLK;
        this.from = from;
        this.tupleBlk = tupleBlk;
    }

    public String getJoinId() {
        return joinId;
    }

    public Type getType() {
        return type;
    }

    public Direction from() {
        return from;
    }

    public int getWindowSize() {
        return windowSize;
    }

    public Tuple getTuple() {
        return tuple;
    }

    public TupleBlock getTupleBlock() {
        return tupleBlk;
    }

    public String toString() {
        StringBuilder sb = new StringBuilder();

        switch (from) {
        case FROM_LEFT:
            sb.append("(-->) ");
            break;

        case FROM_RIGHT:
            sb.append("(<--) ");
            break;

        default:
            sb.append("(?) ");
            break;
        }

        switch (type) {
        case TUPLE:
            sb.append("tuple: [" + tuple + "]");
            break;

        case ACK:
            sb.append("ack: windowSize=" + windowSize);
            break;

        case SIZE_CHG:
            sb.append("size_chg: windowSize=" + windowSize);
            break;

        case TUPLE_BLK:
            sb.append("tuple_blk:\n" + tupleBlk);
            break;

        default:
            sb.append("unknown");
            break;
        }

        return sb.toString();
    }
}